package org.geekbang.projects.messaging;

import org.geektimes.base.util.ReadWriteLock;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * 消息上下文
 *
 * @author : KangNing Hu
 */
public class StreamContext<T> implements Publisher<T> {


	private ReadWriteLock readWriteLock = new ReadWriteLock();

	//所有的订阅者对象
	private List<Subscriber<T>> subscribers = new ArrayList<>();

	//订阅执行者使用线程池
	private Executor executor;

	//是否异步
	private boolean async;


	public StreamContext() {
	}

	public StreamContext(boolean isAsync) {
		this(null, isAsync);
	}

	public StreamContext(Executor executor) {
		this(executor, false);
	}

	public StreamContext(Executor executor, boolean isAsync) {
		this.executor = executor;
		this.async = isAsync;
	}

	@Override
	public void subscribe(Subscriber<? super T> subscriber) {
		SubscriptionDelegate<T> subscriptionDelegate = new SubscriptionDelegate<>(subscriber, this);
		readWriteLock.writeExecute(() -> this.subscribers.add(subscriptionDelegate));
		doSubscribes(subscriptionDelegate);
	}

	/**
	 * 执行订阅
	 */
	private void doSubscribes(SubscriptionDelegate subscriptionDelegate) {
		subscriptionDelegate.onSubscribe(subscriptionDelegate);
	}

	/**
	 * 取消订阅
	 *
	 * @param subscriber
	 */
	public void unsubscribe(Subscriber<? super T> subscriber) {
		readWriteLock.writeExecute(() -> {
			this.subscribers.remove(subscriber);
			return null;
		});
	}



	/**
	 * 发布消息
	 *
	 * @param o
	 */
	public void push(T o) {
		List<Subscriber<T>> subscribers = readWriteLock.readExecute(() -> new ArrayList<>(this.subscribers));
		for (Subscriber<T> subscriber : subscribers){
			if (((SubscriptionDelegate)subscriber).isCancelState()){
				throw new IllegalStateException(String.format("当前订阅[%s]已取消" , subscriber.getClass().getName()));
			}
			doPush(subscriber , o);
		}
	}

	/**
	 * 执行push操作
	 *
	 * @param subscriber
	 * @param o
	 */
	private void doPush(Subscriber<T> subscriber, T o) {
		//异步操作
		if (this.async) {
			this.executor = this.executor == null ? createDefaultExecutor() : this.executor;
			this.executor.execute(() -> {
				subscriber.onNext(o);
			});
		}
		//同步操作
		else {
			subscriber.onNext(o);
		}
	}

	/**
	 * 创建默认线程池
	 *
	 * @return
	 */
	private Executor createDefaultExecutor() {
		return Executors.newCachedThreadPool();
	}
}
